package com.blue.rpc.client;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.blue.rpc.core.message.RpcRequest;
import com.blue.rpc.core.message.RpcResponse;
import com.blue.rpc.util.RpcConst;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class TcpClientHandler extends SimpleChannelInboundHandler<RpcResponse> implements RpcClientHandler
{
	private static Logger logger = LoggerFactory.getLogger(TcpClientHandler.class);

	private final ConcurrentMap<String, RpcClientFutureImpl> futureCache = new ConcurrentHashMap<>();
	private final NettyRpcClient rpcClient;
	private final String token;
	private Channel channel;
	private String host;

	public TcpClientHandler(NettyRpcClient rpcClient, String token)
	{
		this.rpcClient = rpcClient;
		this.token = token;
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception
	{
		if (response.getType() == RpcConst.SHAKE) // 握手报文
		{
			if (response.getRequestId() != null && response.getRequestId().equals(token)) // 验证成功
			{
				logger.info("握手成功，服务端：{}，token：{}", host, token);
				rpcClient.put(host, this);
			}
			else
			{
				logger.warn("握手失败，服务端：{}，本地 token：{}，远程 token：{}", host, token, response.getRequestId());
				//this.close();
			}
		}
		else
		{
			String requestId = response.getRequestId();
			RpcClientFutureImpl future = futureCache.get(requestId);
			if (future != null)
			{
				futureCache.remove(requestId);
				future.done(response);
			}
		}
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception
	{
		super.channelActive(ctx);
		channel = ctx.channel();
		InetSocketAddress remote = (InetSocketAddress)channel.remoteAddress();
		host = String.format(RpcConst.HOST_TPL, remote.getAddress().getHostAddress(), remote.getPort());
		
		// 发送握手报文
		RpcRequest request = new RpcRequest();
		request.setType(RpcConst.SHAKE);
		request.setRequestId(token);
		ctx.writeAndFlush(request);
		logger.info("发送握手报文：{}", token);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception
	{
		super.channelInactive(ctx);
		rpcClient.remove(host);
		//rpcClient.start();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
	{
		logger.error("有错误－", cause);
		ctx.flush().close();
	}
	
	public RpcClientFuture invoke(RpcRequest request)
	{
		RpcClientFutureImpl future = new RpcClientFutureImpl(request);
		futureCache.put(request.getRequestId(), future);
		channel.writeAndFlush(request);
		return future;
	}
	
	public RpcClientFuture invoke(RpcRequest request, RpcClientCallback callback)
	{
		RpcClientFutureImpl future = new RpcClientFutureImpl(request, callback);
		futureCache.put(request.getRequestId(), future);
		channel.writeAndFlush(request);
		return future;
	}
	
	public String getId()
	{
		return channel.id().asLongText();
	}
	
	public void close()
	{
		channel.flush().close();
	}

}


